#include "sock_app.h"
#include "net_io_manager.h"
#include "server_node.h"
#include "util/File.h"
#include "user_session.h"

CNetIOManager::CNetIOManager()
{
    m_event_quit = qcutil::CreateEvent();
    m_event_io = qcutil::CreateEvent();
    m_is_stopped = true;
}
CNetIOManager::~CNetIOManager()
{
    qcutil::DestroyEvent(m_event_quit);
    qcutil::DestroyEvent(m_event_io);
    m_events.clear();
}

void *CNetIOManager::netIOThreadFunction(void* lp)
{
    CNetIOManager* pObj = (CNetIOManager*)lp;
    pObj->netIOThreadImpl();
    return nullptr;
}

std::shared_ptr<IUserSession> CNetIOManager::find_session(std::shared_ptr<CServerNode> node){
    std::lock_guard<std::recursive_mutex> l(m_sessions_mutex);
    auto it = m_sessions.find(node);
    if (it == m_sessions.end()){
        return std::shared_ptr<IUserSession>();
    } else {
        return it->second;
    }
}
void CNetIOManager::set_session(std::shared_ptr<CServerNode> node, std::shared_ptr<IUserSession> session) {
    std::lock_guard<std::recursive_mutex> l(m_sessions_mutex);
    m_sessions[node] = session;
}
void CNetIOManager::del_session(std::shared_ptr<CServerNode> node) {
    std::lock_guard<std::recursive_mutex> l(m_sessions_mutex);
    m_sessions.erase(node);
}

// 确保IO操作都在这个线程中执行。
void CNetIOManager::netIOThreadImpl()
{
    qcutil::neosmart_event_t evts[2] = {0};
    evts[0] = m_event_quit;
    evts[1] = m_event_io;

    while (true)
    {
        int index = -1;
        int x = qcutil::WaitForMultipleEvents(evts, sizeof(evts)/sizeof(evts[0]), false, 1000, index);
        if (x == WAIT_TIMEOUT){
            continue;
        }
        else if (x != 0)
        {
            printf("A wrong value %d returned from function WaitForMultipleEvents\n", x);
            break;
        }
        else
        {
            if (index == WAIT_OBJECT_0)
            {
                //Quit event.
                printf("CNetIOManager::netIOThreadImpl() | Quit event was triggered.\n");
                break;
            }
            else if (index == WAIT_OBJECT_0 + 1)
            {
                // 连接数不多的情况，直接遍历就好。如:
                //  for(i : m_connections) { i->DoRead(); }
                // 如果连接数超过一定数量，就要在设置事件的时候同时告知哪些node(fd)可读。
                printf("CNetIOManager::netIOThreadImpl() | IO event was triggered.\n");
                std::unordered_map<std::shared_ptr<CServerNode>, EventMask> p;
                {
                    std::lock_guard<std::recursive_mutex> l(m_EventsMutex);
                    p.swap(m_events);
                }
                printf("CNetIOManager::netIOThreadImpl() | Event mapping count:%ld\n", p.size());
                for (auto i : p)
                {
                    std::shared_ptr<CServerNode> pNode = i.first;
                    if (!pNode){
                        printf("CNetIOManager::netIOThreadImpl() | Error! connection node cannot be NULL\n");
                        continue;
                    }

                    if (i.second & EM_READ_EVENT)
                    {
                        string recved_data;
                        pNode->getDataRecved(recved_data);
                        std::shared_ptr<IUserSession> ps = find_session(pNode);
                        if (!ps){
                            //@TODO: need a factory here.
                            ps = std::shared_ptr<IUserSession>(new CUserSession(pNode));
                            set_session(pNode, ps);
                        }
                        ps->onUserData(recved_data);
                    }

                    if (i.second & EM_WRITE_EVENT)
                    {
                        printf("处理数据已发送的消息。\n");
                    }

                    if (i.second & EM_ERROR_EVENT)
                    {
                        // 发生了错误。
                        // 发生错误时socket可能已经被close并且此对象已经被从内部列表移除。
                        // 但因为我们这里m_events和m_sessions各持有一个node的智能指针。
                        // 因此在node被从这两个列表中移除前它不会被销毁。
                        printf("CNetIOManager::netIOThreadImpl() | A error event triggered.\n");
                        // @TODO 这里应考虑做更多的事情，或许业务逻辑会对这个连接上发生的错误感兴趣?
                    }

                    if (i.second & EM_CLOSE_EVENT)
                    {
                        printf("CNetIOManager::netIOThreadImpl() | socket closed 事件触发。\n");
                        pNode->closeSocket();
                        // 下面这段不需要了，m_events已经被上面的swap操作清空了。
                        // 考虑在处理这段代码时有新的和pNode的事件写入，那也会在下一次的WaitForMultipleEvents处理
                        // 逻辑中被清空（除非我们的事件处理逻辑有问题否则不用考虑在m_events中清理node的事情）。
                        //{
                        //    std::lock_guard<std::recursive_mutex> l(m_EventsMutex);
                        //    m_events.erase(pNode);
                        //}
                        auto sess = find_session(pNode);
                        if (sess)
                            sess->onSocketClosed();
                        del_session(pNode);
                    }
                }
                p.clear();
            }
            else
            {
                printf("CNetIOManager::netIOThreadImpl() | unknown evnet index %d.\n", index);
                break;
            }
        }
    }
}

void CNetIOManager::start()
{
    //@TODO: 在这里创建处理数据包的线程，和客户端有关的数据包都已经
    //放在对应的CServerNode对象中。我们将它们取出来（收）和放进去（发）。
    m_rsThread.create(netIOThreadFunction, this);
    m_is_stopped = false;
}

void CNetIOManager::stop()
{
    m_is_stopped = true;
    printf("CNetIOManager::stop | set quit event.\n");
    qcutil::SetEvent(m_event_quit);
    m_rsThread.join();
    m_events.clear();
    m_sessions.clear();
}

void CNetIOManager::newSocketAccepted(std::shared_ptr<CServerNode> node)
{
    if (m_is_stopped)
        return;
}

void CNetIOManager::notifySocketClosed(std::shared_ptr<CServerNode> node)
{
    if (m_is_stopped)
        return;
    printf("CNetIOManager::notifySocketClosed() | ..\n");
    {
        std::lock_guard<std::recursive_mutex> l(m_EventsMutex);
        auto it = m_events.find(node);
        if (it != m_events.end())
            it->second |= EM_CLOSE_EVENT;
        else
            m_events[node] = EM_CLOSE_EVENT;
    }
    qcutil::SetEvent(m_event_io);
}

void CNetIOManager::notifySocketError(std::shared_ptr<CServerNode> node, int error_code)
{
    if (m_is_stopped)
        return;
    printf("CNetIOManager::notifySocketError() | ..\n");
    {
        std::lock_guard<std::recursive_mutex> l(m_EventsMutex);
        auto it = m_events.find(node);
        if (it != m_events.end())
            it->second |= EM_ERROR_EVENT;
        else
            m_events[node] = EM_ERROR_EVENT;
    }
    qcutil::SetEvent(m_event_io);
}

void CNetIOManager::notifyDataRecvedEvent(std::shared_ptr<CServerNode> node)
{
    if (m_is_stopped){
        printf("CNetIOManager::notifyDataRecvedEvent | IO Manager was stopped.\n");
        return;
    }
    // 通知数据处理线程，在一个连接上已有内容被接收。
    // 即CServerNode中的工作线程已经将数据缓存到其数据接收缓存m_recv_buffers中。
    // 在设计上做了分离，数据接收和数据处理分别用隶属不同对象（即业务层和网络层）的两个线程处理，以达到异步的目的。
    {
        std::lock_guard<std::recursive_mutex> l(m_EventsMutex);
        auto it = m_events.find(node);
        if (it != m_events.end())
            it->second |= EM_READ_EVENT;
        else
            m_events[node] = EM_READ_EVENT;
    }
    printf("CNetIOManager::notifyDataRecvedEvent() | Set IO event.\n");
    qcutil::SetEvent(m_event_io);
}

void CNetIOManager::notifyDataSentEvent(std::shared_ptr<CServerNode> node)
{
    if (m_is_stopped)
    {
        printf("CNetIOManager::notifyDataSentEvent() | IO Manager was stopped.\n");
        return;
    }
    // 通知数据处理线程，在一个连接上已有内容被发送。
    // 即CServerNode中的工作线程已经将数据缓存到其数据发送缓存中。
    {
        std::lock_guard<std::recursive_mutex> l(m_EventsMutex);
        auto it = m_events.find(node);
        if (it != m_events.end())
            it->second |= EM_WRITE_EVENT;
        else
            m_events[node] = EM_WRITE_EVENT;
    }
    
    printf("CNetIOManager::notifyDataSentEvent() | Set IO event.\n");
    qcutil::SetEvent(m_event_io);
}
